package cn.z2huo.demo.kafka.spring.retry.nonblock.config.cluster1;

import cn.z2huo.demo.kafka.spring.retry.nonblock.properties.KafkaConfiguration;
import cn.z2huo.demo.kafka.spring.retry.nonblock.support.Cluster1ProducerListener;
import lombok.RequiredArgsConstructor;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;

/**
 * <p>
 *
 * @author z2huo
 */
@EnableConfigurationProperties(KafkaConfiguration.class)
@Configuration
@RequiredArgsConstructor
public class KafkaProducerConfig {

    private final KafkaConfiguration kafkaConfiguration;

    private final Cluster1ProducerListener producerListener;

    @Bean(name = "cluster1KafkaProducerFactory")
    public ProducerFactory<String, String> kafkaProducerFactory() {
        DefaultKafkaProducerFactory<String, String> factory =
                new DefaultKafkaProducerFactory<>(kafkaConfiguration.getCluster1().buildProducerProperties());
        return factory;
    }

    @Bean(name = "cluster1KafkaTemplate")
    public KafkaTemplate<String, String> testTopicKafkaTemplate(
            @Qualifier("cluster1KafkaProducerFactory") ProducerFactory<String, String> producerFactory
    ) {
        KafkaTemplate<String, String> kafkaTemplate = new KafkaTemplate<>(producerFactory);
        kafkaTemplate.setProducerListener(producerListener);
        return kafkaTemplate;
    }

}
